package cn.dglydrpy.study.j2ee.kafka.producer;

import org.apache.kafka.clients.producer.*;
import org.apache.kafka.common.serialization.StringSerializer;

import cn.dglydrpy.study.j2ee.kafka.entity.User;
import cn.dglydrpy.study.j2ee.kafka.serialization.UserSerializer;

import java.util.HashMap;
import java.util.Map;
/**
 * Kafka使用接口org.apache.kafka.common.serialization.Serializer 定义序列化器，将泛型指定类型的数据转换为字节数组
 * 
 * 自带的实现：
 * ByteArraySerializer
 * ByteBufferSerializer
 * BytesSerializer
 * DoubleSerializer
 * FloatSerializer
 * IntegerSerializer
 * StringSerializer
 * LongSerializer
 * ShortSerializer
 * 可通过实现Serializer自定义序列化器
 * 
 * 数据的序列化一般生产中使用avro。
 * 
 * @author LY_飘涯
 *
 */
public class MyKafkaUserProducer {
    public static void main(String[] args) {
        Map<String, Object> configs = new HashMap<>();
        configs.put(ProducerConfig.BOOTSTRAP_SERVERS_CONFIG, "192.168.80.131:9092");
        configs.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, StringSerializer.class);
        // 设置自定义的序列化器
        configs.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, UserSerializer.class);

        KafkaProducer<String, User> producer = new KafkaProducer<String, User>(configs);

        User user = new User();
        user.setUserId(1005);
//        user.setUsername("张三");
//        user.setUsername("李四");
        user.setUsername("王五");
//        user.setUserId(400);
//        user.setUsername("赵四");

        // 需先创建tp_user主题
        ProducerRecord<String, User> record = new ProducerRecord<String, User>(
                "tp_user",   // topic
                user.getUsername(),   // key
                user                  // value
        );


        producer.send(record, new Callback() {
            @Override
            public void onCompletion(RecordMetadata metadata, Exception exception) {
                if (exception != null) {
                    System.out.println("消息发送异常");
                    exception.printStackTrace();
                } else {
                    System.out.println("主题：" + metadata.topic() + "\t"
                    + "分区：" + metadata.partition() + "\t"
                    + "生产者偏移量：" + metadata.offset());
                }
            }
        });

        // 关闭生产者
        producer.close();

    }
}
